package com.xdxc.service.impl;


import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.xdxc.config.GradingConfig;
import com.xdxc.exception.GradingException;
import com.xdxc.feign.ExamFeign;
import com.xdxc.model.entity.TempUserAnswer;
import com.xdxc.service.serviceinterface.GradingResultService;
import com.xdxc.service.serviceinterface.ProcessedBatchService;
import com.xdxc.service.serviceinterface.TempUserAnswerService;
import com.xdxc.utils.SecurityUtils;
import com.xdxc.utils.TokenHolder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.security.concurrent.DelegatingSecurityContextExecutorService;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;


@Slf4j
@Service
public class GradingService {

    // 配置参数（带默认值）
    @Value("${grading.pageSize:1000}")
    private int pageSize;

    @Value("${grading.maxRetry:3}")
    private int maxRetry;

    @Value("${grading.maxConcurrent:5}")
    private int maxConcurrent;

    @Value("${grading.rateLimit:100}")
    private int rateLimit;

    @Value("${grading.scoringBatchSize:100}")
    private int scoringBatchSize;

    @Autowired
    private TempUserAnswerService tempUserAnswerService;
    @Autowired
    private ExamFeign examFeign;


    @Autowired
    private ProcessedBatchService processedBatchService;

    @Autowired
    private GradingResultService gradingResultService;
    @Autowired
    private GradingConfig config;
    // 线程池配置
    private ExecutorService gradingExecutor;
    @PostConstruct
    public void init() {
        // 1. 创建原生线程池
        ThreadPoolExecutor rawExecutor = new ThreadPoolExecutor(
                maxConcurrent,
                maxConcurrent,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(maxConcurrent * 2),
                new GradingThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        // 2. 包装成支持SecurityContext的线程池
        gradingExecutor = new DelegatingSecurityContextExecutorService(
                rawExecutor,
                SecurityContextHolder.getContext() // 自动传递当前安全上下文
        );
    }
    @PreDestroy
    public void shutdown() {
        if (gradingExecutor != null) {
            gradingExecutor.shutdownNow();
        }
    }
    @Transactional(rollbackFor = Exception.class)
    public void sendExamAnswers(Long examId) {
        validateExamId(examId);
        final RateLimiter rateLimiter = RateLimiter.create(rateLimit);
        final AtomicInteger currentPage = new AtomicInteger(1);
        final Semaphore semaphore = new Semaphore(maxConcurrent);
        final List<CompletableFuture<Void>> futures = new ArrayList<>();
        try {
            Long lastId = getInitialLastId(examId);
            while (true) {
                rateLimiter.acquire();
                final int currentPageNum = currentPage.get();
                List<TempUserAnswer> batch = queryBatch(examId, lastId);
                if (CollectionUtils.isEmpty(batch)) {
                    break;
                }
                lastId = batch.get(batch.size() - 1).getId();

                semaphore.acquire();
                final List<TempUserAnswer> finalBatch = batch;
                // 更新lastId为当前批次的最后一个ID
                String token = SecurityUtils.getCurrentToken();
                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                    TokenHolder.setToken(token);
                    try {
                        processBatchWithMetrics(examId, currentPageNum, finalBatch);
                    } finally {
                        semaphore.release();
                    }
                }, gradingExecutor).exceptionally(e -> {
                    handleProcessingError(examId, currentPageNum, e);
                    return null;
                });
                futures.add(future);
                currentPage.incrementAndGet();
            }
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
            log.info("Exam processing completed: examId={}, totalPages={}", examId, currentPage.get()-1);

        } catch (Exception e) {

            throw new GradingException("Exam processing failed: examId=" + examId, e);
        }
    }
    // 获取初始lastId的方法
    private Long getInitialLastId(Long examId) {
        // 方式1：从数据库查询最小ID（确保不遗漏任何记录）
        LambdaQueryWrapper<TempUserAnswer> query = new LambdaQueryWrapper<>();
        query.select(TempUserAnswer::getId)
                .eq(TempUserAnswer::getExamId, examId)
                .orderByAsc(TempUserAnswer::getId)
                .last("LIMIT 1");
        TempUserAnswer firstRecord = tempUserAnswerService.getOne(query);
        return firstRecord != null ? firstRecord.getId() - 1 : 0L; // 从第一个记录开始

        // 或者方式2：如果支持游标分页，可以使用特定起始值
        // return config.getInitialLastId(); // 从配置获取
    }

    private List<TempUserAnswer> queryBatch(Long examId, Long lastId) {
        LambdaQueryWrapper<TempUserAnswer> query = new LambdaQueryWrapper<>();
        query.eq(TempUserAnswer::getExamId, examId)
                .gt(TempUserAnswer::getId, lastId)
                .ne(TempUserAnswer::getIsEnd, 1)  // 添加 isEnd != 1 的条件
                .orderByAsc(TempUserAnswer::getId)
                .last("LIMIT " + config.getPageSize());
        return tempUserAnswerService.list(query);
    }
    private void processBatchWithMetrics(Long examId, int batchNum, List<TempUserAnswer> batch) {
        try {
            int retryCount = 0;
            while (retryCount <= maxRetry) {
                try {
                    processBatch(examId, batch);
                    return;
                } catch (Exception e) {
                    retryCount++;
                    if (retryCount > maxRetry) {
                        throw new GradingException(String.format(
                                "Batch processing failed after %d retries: examId=%d, batch=%d",
                                maxRetry, examId, batchNum), e);
                    }
                    applyBackoff(retryCount);
                }
            }
        } catch (Exception e) {
            throw e;
        }
        finally {
            if (gradingExecutor != null) {
                gradingExecutor.shutdownNow();
            }
        }
    }

    private void processBatch(Long examId, List<TempUserAnswer> batch) {
        String batchId = generateBatchId(examId, batch);
        if (processedBatchService.isProcessed(batchId)) {
            log.debug("Skipping processed batch: {}", batchId);
            return;
        }

        long startTime = System.currentTimeMillis();
        try {
            processInParallel(batch, examId);
            processedBatchService.markProcessed(batchId, examId);
            tempUserAnswerService
                    .update(
                    new LambdaUpdateWrapper<TempUserAnswer>().in(TempUserAnswer::getId, batch
                        .stream()
                            .map(TempUserAnswer::getId) // 提取 id 字段
                            .collect(Collectors.toList()))  // WHERE id IN (idList)
                    .set(TempUserAnswer::getIsEnd, 1));
            log.debug("Batch processed successfully: batchId={}, time={}ms",
                    batchId, System.currentTimeMillis()-startTime);
        } catch (Exception e) {
            log.error("Batch processing failed: batchId={}", batchId, e);
            throw new GradingException("Batch processing error", e);
        }
    }

    private void processInParallel(List<TempUserAnswer> batch, Long examId) {
        // 使用并行流+分片处理
        int numBatches = (int) Math.ceil((double) batch.size() / scoringBatchSize);
        IntStream.range(0, numBatches)
                .parallel()
                .forEach(batchIndex -> {
                    int fromIndex = batchIndex * scoringBatchSize;
                    int toIndex = Math.min(fromIndex + scoringBatchSize, batch.size());
                    List<TempUserAnswer> subBatch = batch.subList(fromIndex, toIndex);

                    try {
                        gradingResultService.objectGrade(subBatch, examId);
                    } catch (Exception e) {
                        throw new GradingException(String.format(
                                "Sub-batch grading failed: examId=%d, batchIndex=%d",
                                examId, batchIndex), e);
                    }
                });
    }

    // 其他工具方法
    private LambdaQueryWrapper<TempUserAnswer> buildBaseQuery(Long examId) {
        return new LambdaQueryWrapper<TempUserAnswer>()
                .eq(TempUserAnswer::getExamId, examId)
                .orderByAsc(TempUserAnswer::getId); // 确保分页稳定性
    }

    private String generateBatchId(Long examId, List<TempUserAnswer> batch) {
        String idSequence = batch.stream()
                .map(answer -> answer.getId().toString())
                .collect(Collectors.joining("|"));
        return examId + "_" + DigestUtils.sha256Hex(idSequence);
    }

    private void applyBackoff(int retryCount) {
        try {
            long delay = (long) Math.pow(2, retryCount) * 1000 + ThreadLocalRandom.current().nextInt(500);
            TimeUnit.MILLISECONDS.sleep(delay);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new GradingException("Retry interrupted", e);
        }
    }

    private void validateExamId(Long examId) {
        if (examId == null || examId <= 0) {
            throw new IllegalArgumentException("Invalid exam ID: " + examId);
        }


    }

    private boolean isLastPage(IPage<?> page, int currentPage) {
        return currentPage >= page.getPages() || page.getRecords().size() < pageSize;
    }

    private void handleProcessingError(Long examId, int batchNum, Throwable e) {
        log.error("Critical processing failure: examId={}, batch={}", examId, batchNum, e);
        throw new GradingException("Critical processing error", e);
    }

    // 自定义线程工厂
    private static class GradingThreadFactory implements ThreadFactory {
        private final AtomicInteger threadNumber = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r, "grading-pool-" + threadNumber.getAndIncrement());
            thread.setDaemon(true);
            thread.setPriority(Thread.NORM_PRIORITY);
            return thread;
        }
    }
}

// 自定义异常类




